package com.mall.rabbitmq.demo2;

import com.rabbitmq.client.*;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

@Component
public class Receiver {
    private final static String QUEUE_NAME = "hello";
    private static final String TASK_QUEUE_NAME = "task_queue";
    private static final String EXCHANGE_NAME = "logs";
    private static final String EXCHANGE_NAME2 = "direct_logs";

    public static void main(String[] argv) throws Exception {
        routePattern();
    }
    //普通
    public static void demo() throws Exception {
        //创建链接工厂对象
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //设置RabbitMQ服务主机地址,默认localhost
        connectionFactory.setHost("localhost");
        //设置RabbitMQ服务端口,默认5672
        connectionFactory.setPort(5672);
        //设置虚拟主机名字(rabbitmq服务器上创建的虚拟主机)
        connectionFactory.setVirtualHost("demo");
        //设置用户连接名，默认guest(指定用哪个用户登录，guest是默认超级管理员)
        connectionFactory.setUsername("guest");
        //设置链接密码，默认guest
        connectionFactory.setPassword("guest");
        //创建链接(连接到rabbitmq服务器)
        Connection connection = connectionFactory.newConnection();
        //创建频道(创建连接rabbitmq服务器之间的稳定、高效的频道，持久通讯）
        Channel channel = connection.createChannel();
        //申明队列，指定到哪个队列获取消息
        /**
         * 声明队列
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接(其它连接是否能连接到本条队列)
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         * **/
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //创建消费者，并设置消息处理(DefaultConsumer:消息消费者，参数传入创建的频道)然后再重写handleDelivery方法，可以用lambab表达式
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            /***
             * @param consumerTag   消息者标签，在channel.basicConsume时候可以指定
             * @param envelope      消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * @param properties    属性信息
             * @param body           消息
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由的key
                String routingKey = envelope.getRoutingKey();
                //获取交换机信息
                String exchange = envelope.getExchange();
                //获取消息ID
                long deliveryTag = envelope.getDeliveryTag();
                //获取消息信息
                String message = new String(body, "UTF-8");
                //输出获得的消息内容
                System.out.println("routingKey:"+routingKey+",exchange:"+exchange+",deliveryTag:"+deliveryTag+",message:"+message);
            }
        };
        //消息监听
        /**
         * 消息监听
         * 要监听哪个队列？当消费者收到消息之后是否自动告诉rebbitmq服务器已经收到？收到消息之后，如何处理呢？
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调(传入上面创建的消费者对象，这个消费者对象中对做了对收到的消息处理)
         */
        channel.basicConsume(QUEUE_NAME,true,defaultConsumer);
        //关闭资源(不建议关闭，建议一直监听消息)
        //channel.close();
        //connection.close();
    }

    //worker模式
    public static void workerPattern() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        final Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();
        /**
         * 声明队列
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接(其它连接是否能连接到本条队列)
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         * **/
        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        //可以把预取的初始数量设置为1,这样,性能较高的消费者就会去消费更过的消息,大大提高效率
        //channel.basicQos(1);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                //channel.basicAck()方法在消息消费完成时通知服务端消费已经完成。
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
            System.out.println(" [x] Received '" + message + "'");
        };
        //手动确认模式 消费者从队列中获取消息之后，服务器会将该消息标记为不可用状态，等待消费者的反馈，如果消费者一直没有反馈，那么该消息将一直处于不可用状态。
        //手动设置需要使用 channel.basicAck() 方法来通知服务端消费已完成
        channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> { });
    }


    //Pub/Sub模式
    public static void pubPattern() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        /**
         * 声明交换机
         * 参数1：交换机名称
         * 参数2：交换机类型，fanout、topic、direct、headers(以下用fanout类型，广播模式，每个与交换机绑定的队列都会接收到信息)
         */
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        //定义一个匿名额队列
        String queueName = channel.queueDeclare().getQueue();
        //将队列和交换机绑定
        channel.queueBind(queueName, EXCHANGE_NAME, "");

        System.out.println(" [*] Waiting for messages. To exit press CTRL+C"+queueName);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
        };
        /**
         * 消息监听
         * 要监听哪个队列？当消费者收到消息之后是否自动告诉rebbitmq服务器已经收到？收到消息之后，如何处理呢？
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调(传入上面创建的消费者对象，这个消费者对象中对做了对收到的消息处理)
         */
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
    }

    //路由模式
    public static void routePattern() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME2, BuiltinExchangeType.DIRECT);
        String queueName = channel.queueDeclare().getQueue();

        channel.queueBind(queueName, EXCHANGE_NAME2, "info");
        channel.queueBind(queueName, EXCHANGE_NAME2, "error");
        channel.queueBind(queueName, EXCHANGE_NAME2, "warning");
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C"+queueName);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
        });
    }

}
